//! # 消息总线核心抽象
//!
//! 定义消息总线系统的核心 trait 和接口

use async_trait::async_trait;
use std::sync::Arc;
use tokio::sync::mpsc;

use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;

/// 消息总线核心接口
#[async_trait]
pub trait MessageBus: Send + Sync {
    /// 发布消息到指定主题
    ///
    /// # 参数
    /// - `message`: 要发布的消息
    ///
    /// # 返回
    /// - `Ok(())`: 发布成功
    /// - `Err(MessageBusError)`: 发布失败
    async fn publish(&self, message: Message) -> MessageBusResult<()>;

    /// 订阅主题，返回订阅者
    ///
    /// # 参数
    /// - `pattern`: 主题模式，支持通配符（如 "user.events.*"）
    ///
    /// # 返回
    /// - `Ok(Subscriber)`: 订阅成功，返回订阅者
    /// - `Err(MessageBusError)`: 订阅失败
    async fn subscribe(&self, pattern: &str) -> MessageBusResult<Box<dyn Subscriber>>;

    /// 取消订阅
    ///
    /// # 参数
    /// - `subscription_id`: 订阅 ID
    ///
    /// # 返回
    /// - `Ok(())`: 取消订阅成功
    /// - `Err(MessageBusError)`: 取消订阅失败
    async fn unsubscribe(&self, subscription_id: &str) -> MessageBusResult<()>;

    /// 获取主题的订阅者数量
    ///
    /// # 参数
    /// - `topic`: 主题名称
    ///
    /// # 返回
    /// - `Ok(usize)`: 订阅者数量
    /// - `Err(MessageBusError)`: 获取失败
    async fn subscriber_count(&self, topic: &str) -> MessageBusResult<usize>;

    /// 获取所有活跃主题列表
    ///
    /// # 返回
    /// - `Ok(Vec<String>)`: 主题列表
    /// - `Err(MessageBusError)`: 获取失败
    async fn list_topics(&self) -> MessageBusResult<Vec<String>>;

    /// 检查主题是否存在订阅者
    ///
    /// # 参数
    /// - `topic`: 主题名称
    ///
    /// # 返回
    /// - `Ok(bool)`: 是否存在订阅者
    /// - `Err(MessageBusError)`: 检查失败
    async fn has_subscribers(&self, topic: &str) -> MessageBusResult<bool>;

    /// 关闭消息总线
    ///
    /// # 返回
    /// - `Ok(())`: 关闭成功
    /// - `Err(MessageBusError)`: 关闭失败
    async fn close(&self) -> MessageBusResult<()>;

    /// 获取消息总线统计信息
    ///
    /// # 返回
    /// - `Ok(MessageBusStats)`: 统计信息
    /// - `Err(MessageBusError)`: 获取失败
    async fn stats(&self) -> MessageBusResult<MessageBusStats>;
}

/// 消息发布者接口
#[async_trait]
pub trait Publisher: Send + Sync {
    /// 发布消息
    ///
    /// # 参数
    /// - `message`: 要发布的消息
    ///
    /// # 返回
    /// - `Ok(())`: 发布成功
    /// - `Err(MessageBusError)`: 发布失败
    async fn publish(&self, message: Message) -> MessageBusResult<()>;

    /// 批量发布消息
    ///
    /// # 参数
    /// - `messages`: 要发布的消息列表
    ///
    /// # 返回
    /// - `Ok(())`: 全部发布成功
    /// - `Err(MessageBusError)`: 发布失败
    async fn publish_batch(&self, messages: Vec<Message>) -> MessageBusResult<()>;

    /// 发布并等待确认
    ///
    /// # 参数
    /// - `message`: 要发布的消息
    ///
    /// # 返回
    /// - `Ok(())`: 发布并确认成功
    /// - `Err(MessageBusError)`: 发布或确认失败
    async fn publish_and_confirm(&self, message: Message) -> MessageBusResult<()>;

    /// 关闭发布者
    async fn close(&self) -> MessageBusResult<()>;
}

/// 消息订阅者接口
#[async_trait]
pub trait Subscriber: Send + Sync {
    /// 接收消息（非阻塞）
    ///
    /// # 返回
    /// - `Ok(Some(Message))`: 接收到消息
    /// - `Ok(None)`: 没有消息
    /// - `Err(MessageBusError)`: 接收失败
    async fn try_receive(&self) -> MessageBusResult<Option<Message>>;

    /// 接收消息（阻塞）
    ///
    /// # 返回
    /// - `Ok(Some(Message))`: 接收到消息
    /// - `Ok(None)`: 订阅者关闭
    /// - `Err(MessageBusError)`: 接收失败
    async fn receive(&self) -> MessageBusResult<Option<Message>>;

    /// 接收消息（带超时）
    ///
    /// # 参数
    /// - `timeout_ms`: 超时时间（毫秒）
    ///
    /// # 返回
    /// - `Ok(Some(Message))`: 接收到消息
    /// - `Ok(None)`: 超时或订阅者关闭
    /// - `Err(MessageBusError)`: 接收失败
    async fn receive_timeout(&self, timeout_ms: u64) -> MessageBusResult<Option<Message>>;

    /// 确认消息处理完成
    ///
    /// # 参数
    /// - `message`: 要确认的消息
    ///
    /// # 返回
    /// - `Ok(())`: 确认成功
    /// - `Err(MessageBusError)`: 确认失败
    async fn ack(&self, message: &Message) -> MessageBusResult<()>;

    /// 拒绝消息（重新入队或丢弃）
    ///
    /// # 参数
    /// - `message`: 要拒绝的消息
    /// - `requeue`: 是否重新入队
    ///
    /// # 返回
    /// - `Ok(())`: 拒绝成功
    /// - `Err(MessageBusError)`: 拒绝失败
    async fn nack(&self, message: &Message, requeue: bool) -> MessageBusResult<()>;

    /// 获取订阅 ID
    ///
    /// # 返回
    /// - 订阅 ID
    fn subscription_id(&self) -> &str;

    /// 获取订阅模式
    ///
    /// # 返回
    /// - 订阅模式
    fn pattern(&self) -> &str;

    /// 检查订阅者是否仍然活跃
    ///
    /// # 返回
    /// - 是否活跃
    fn is_active(&self) -> bool;

    /// 关闭订阅者
    async fn close(&self) -> MessageBusResult<()>;

    /// 获取订阅者统计信息
    ///
    /// # 返回
    /// - `Ok(SubscriberStats)`: 统计信息
    /// - `Err(MessageBusError)`: 获取失败
    async fn stats(&self) -> MessageBusResult<SubscriberStats>;
}

/// 消息总线统计信息
#[derive(Debug, Clone, Default)]
pub struct MessageBusStats {
    /// 总发布消息数
    pub total_published: u64,
    /// 总接收消息数
    pub total_received: u64,
    /// 总失败消息数
    pub total_failed: u64,
    /// 当前活跃订阅者数
    pub active_subscribers: usize,
    /// 当前活跃主题数
    pub active_topics: usize,
    /// 消息队列长度
    pub queue_length: usize,
    /// 运行时间（秒）
    pub uptime_seconds: u64,
    /// 每秒消息处理率
    pub messages_per_second: f64,
    /// 平均消息大小（字节）
    pub average_message_size: usize,
    /// 内存使用量（字节）
    pub memory_usage: usize,
}

/// 订阅者统计信息
#[derive(Debug, Clone, Default)]
pub struct SubscriberStats {
    /// 总接收消息数
    pub total_received: u64,
    /// 总确认消息数
    pub total_acked: u64,
    /// 总拒绝消息数
    pub total_nacked: u64,
    /// 当前队列长度
    pub queue_length: usize,
    /// 平均处理时间（毫秒）
    pub average_processing_time_ms: f64,
    /// 最后接收消息时间
    pub last_received_at: Option<chrono::DateTime<chrono::Utc>>,
    /// 订阅开始时间
    pub subscribed_at: chrono::DateTime<chrono::Utc>,
}

/// 消息总线工厂接口
#[async_trait]
pub trait MessageBusFactory: Send + Sync {
    /// 创建消息总线实例
    ///
    /// # 参数
    /// - `config`: 配置参数
    ///
    /// # 返回
    /// - `Ok(Arc<dyn MessageBus>)`: 消息总线实例
    /// - `Err(MessageBusError)`: 创建失败
    async fn create(&self, config: &dyn std::any::Any) -> MessageBusResult<Arc<dyn MessageBus>>;

    /// 获取工厂名称
    ///
    /// # 返回
    /// - 工厂名称
    fn name(&self) -> &str;

    /// 获取支持的配置类型
    ///
    /// # 返回
    /// - 配置类型名称
    fn config_type(&self) -> &str;
}

/// 消息过滤器接口
pub trait MessageFilter: Send + Sync {
    /// 检查消息是否匹配过滤条件
    ///
    /// # 参数
    /// - `message`: 要检查的消息
    ///
    /// # 返回
    /// - `true`: 匹配
    /// - `false`: 不匹配
    fn matches(&self, message: &Message) -> bool;

    /// 获取过滤器描述
    ///
    /// # 返回
    /// - 过滤器描述
    fn description(&self) -> String;
}

/// 消息拦截器接口
#[async_trait]
pub trait MessageInterceptor: Send + Sync {
    /// 在消息发布前拦截
    ///
    /// # 参数
    /// - `message`: 要拦截的消息
    ///
    /// # 返回
    /// - `Ok(Some(Message))`: 继续处理（可能修改过的消息）
    /// - `Ok(None)`: 丢弃消息
    /// - `Err(MessageBusError)`: 拦截失败
    async fn before_publish(&self, message: Message) -> MessageBusResult<Option<Message>>;

    /// 在消息发布后拦截
    ///
    /// # 参数
    /// - `message`: 已发布的消息
    /// - `result`: 发布结果
    async fn after_publish(&self, message: &Message, result: &MessageBusResult<()>);

    /// 在消息接收前拦截
    ///
    /// # 参数
    /// - `message`: 要拦截的消息
    ///
    /// # 返回
    /// - `Ok(Some(Message))`: 继续处理（可能修改过的消息）
    /// - `Ok(None)`: 丢弃消息
    /// - `Err(MessageBusError)`: 拦截失败
    async fn before_receive(&self, message: Message) -> MessageBusResult<Option<Message>>;

    /// 在消息接收后拦截
    ///
    /// # 参数
    /// - `message`: 已接收的消息
    /// - `subscriber_id`: 订阅者 ID
    async fn after_receive(&self, message: &Message, subscriber_id: &str);
}

/// 简单内存订阅者实现
pub struct MemorySubscriber {
    pub subscription_id: String,
    pub pattern: String,
    pub receiver: Arc<tokio::sync::Mutex<mpsc::UnboundedReceiver<Message>>>,
    pub stats: Arc<tokio::sync::Mutex<SubscriberStats>>,
    pub is_active: Arc<std::sync::atomic::AtomicBool>,
}

impl MemorySubscriber {
    pub fn new(
        subscription_id: String,
        pattern: String,
        receiver: mpsc::UnboundedReceiver<Message>,
    ) -> Self {
        Self {
            subscription_id,
            pattern,
            receiver: Arc::new(tokio::sync::Mutex::new(receiver)),
            stats: Arc::new(tokio::sync::Mutex::new(SubscriberStats {
                subscribed_at: chrono::Utc::now(),
                ..Default::default()
            })),
            is_active: Arc::new(std::sync::atomic::AtomicBool::new(true)),
        }
    }
}

#[async_trait]
impl Subscriber for MemorySubscriber {
    async fn try_receive(&self) -> MessageBusResult<Option<Message>> {
        if !self.is_active() {
            return Ok(None);
        }

        let mut receiver = self.receiver.lock().await;
        match receiver.try_recv() {
            Ok(mut message) => {
                message.metadata_mut().mark_received();
                
                let mut stats = self.stats.lock().await;
                stats.total_received += 1;
                stats.last_received_at = Some(chrono::Utc::now());
                
                Ok(Some(message))
            }
            Err(mpsc::error::TryRecvError::Empty) => Ok(None),
            Err(mpsc::error::TryRecvError::Disconnected) => {
                self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
                Ok(None)
            }
        }
    }

    async fn receive(&self) -> MessageBusResult<Option<Message>> {
        if !self.is_active() {
            return Ok(None);
        }

        let mut receiver = self.receiver.lock().await;
        match receiver.recv().await {
            Some(mut message) => {
                message.metadata_mut().mark_received();
                
                let mut stats = self.stats.lock().await;
                stats.total_received += 1;
                stats.last_received_at = Some(chrono::Utc::now());
                
                Ok(Some(message))
            }
            None => {
                self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
                Ok(None)
            }
        }
    }

    async fn receive_timeout(&self, timeout_ms: u64) -> MessageBusResult<Option<Message>> {
        if !self.is_active() {
            return Ok(None);
        }

        let timeout = tokio::time::Duration::from_millis(timeout_ms);
        let mut receiver = self.receiver.lock().await;
        
        match tokio::time::timeout(timeout, receiver.recv()).await {
            Ok(Some(mut message)) => {
                message.metadata_mut().mark_received();
                
                let mut stats = self.stats.lock().await;
                stats.total_received += 1;
                stats.last_received_at = Some(chrono::Utc::now());
                
                Ok(Some(message))
            }
            Ok(None) => {
                self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
                Ok(None)
            }
            Err(_) => Ok(None), // 超时
        }
    }

    async fn ack(&self, _message: &Message) -> MessageBusResult<()> {
        let mut stats = self.stats.lock().await;
        stats.total_acked += 1;
        Ok(())
    }

    async fn nack(&self, _message: &Message, _requeue: bool) -> MessageBusResult<()> {
        let mut stats = self.stats.lock().await;
        stats.total_nacked += 1;
        Ok(())
    }

    fn subscription_id(&self) -> &str {
        &self.subscription_id
    }

    fn pattern(&self) -> &str {
        &self.pattern
    }

    fn is_active(&self) -> bool {
        self.is_active.load(std::sync::atomic::Ordering::Relaxed)
    }

    async fn close(&self) -> MessageBusResult<()> {
        self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
        Ok(())
    }

    async fn stats(&self) -> MessageBusResult<SubscriberStats> {
        let stats = self.stats.lock().await;
        Ok(stats.clone())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::message::Message;
    use serde_json::json;

    #[tokio::test]
    async fn test_memory_subscriber() {
        let (sender, receiver) = mpsc::unbounded_channel();
        let subscriber = MemorySubscriber::new(
            "test-subscription".to_string(),
            "test.*".to_string(),
            receiver,
        );

        assert_eq!(subscriber.subscription_id(), "test-subscription");
        assert_eq!(subscriber.pattern(), "test.*");
        assert!(subscriber.is_active());

        // 发送消息
        let message = Message::new("test.topic", json!({"test": "data"}));
        sender.send(message.clone()).unwrap();

        // 接收消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        assert_eq!(received.unwrap().payload(), message.payload());

        // 测试 ack
        assert!(subscriber.ack(&message).await.is_ok());

        // 测试统计信息
        let stats = subscriber.stats().await.unwrap();
        assert_eq!(stats.total_received, 1);
        assert_eq!(stats.total_acked, 1);
    }
}